// -*- mode:C; tab-width:2; indent-tabs-mode:nil;  -*-

#include <stdlib.h>
#include <string.h>
#include <pthread.h>

#include "lockDef.h"
#include "queue.h"
#include "taskQueue.h"
#include "factory.h"


INSTANTIATETYPE(taskQueue_t, 
                queue_t q; 
                int nWorkers; 
                pthread_t* worker_threads;
                int isDestroyed;
                pthread_mutex_t lock;
                pthread_cond_t append_or_destroy_cond;)
INSTANTIATETYPE(taskDescriptor_t,
                taskQueue_task_fn_t f; 
                taskQueue_closure_t c;
                taskQueue_taskDone_fn_t doneFn;)

CREATEMANAGER(taskQueue_t) 
CREATEMANAGER(taskDescriptor_t)
  
void* taskQueueworkerFn(void* tqp) {
  while (1) {    
    if (!(*(taskQueue_t *)tqp)->nWorkers){//No worker
      taskDescriptor_t td;
      queue_remove((*(taskQueue_t*)tqp)->q, &td);
      if (td->f) {
        taskQueue_taskResult_t* result = (taskQueue_taskResult_t*)malloc(sizeof(taskQueue_taskResult_t));
        *result = td->f(td->c);
        if (td->doneFn){
          td->doneFn(td->c, *result);
        }
        free(result);
        }
      taskDescriptor_t_release(td,NULL);
      return;
    }else{//at least one worker 
      pthread_mutex_lock(&((*(taskQueue_t *)tqp)->lock));
      while(queue_is_empty((*(taskQueue_t *)tqp)->q)&&(*(taskQueue_t*)tqp)->isDestroyed==0){
         //printf("Wait d = %d\n", d);      
         pthread_cond_wait(&((*(taskQueue_t *)tqp)->append_or_destroy_cond), &((*(taskQueue_t *)tqp)->lock));//return FALSE;
      }
       
      if ((*(taskQueue_t*)tqp)->isDestroyed){// destroy is on        
        int size = queue_size((*(taskQueue_t *)tqp)->q);
        if (size==0){
          pthread_mutex_unlock(&((*(taskQueue_t *)tqp)->lock));        
          return 0; //maybe it doesnt exit here
        }        
      }
      taskDescriptor_t td;
      queue_remove((*(taskQueue_t*)tqp)->q, &td);
      if (td->f) {
        taskQueue_taskResult_t* result = (taskQueue_taskResult_t*)malloc(sizeof(taskQueue_taskResult_t));
        *result = td->f(td->c);
        if (td->doneFn){
          td->doneFn(td->c, *result);
        }
        free(result);
      }
      taskDescriptor_t_release(td,NULL);  
      pthread_mutex_unlock(&((*(taskQueue_t *)tqp)->lock));   
      }
  }
}

//tqp already initialized
int taskQueue_create(taskQueue_t* tqp, int nWorkers) {
  int i;
  *tqp = taskQueue_t_acquire();
  pthread_mutex_init(&((*tqp)->lock),NULL);
  (*tqp)->q = queue_create();
  (*tqp)->nWorkers = nWorkers;
  (*tqp)->isDestroyed = 0;
  if (nWorkers){
    (*tqp)->worker_threads = (pthread_t*) malloc(nWorkers*sizeof(pthread_t));
     pthread_cond_init(&((*tqp)->append_or_destroy_cond),NULL);
    for (i=0; i<nWorkers; i++) {
      pthread_create(&(*tqp)->worker_threads[i], NULL, taskQueueworkerFn, tqp);
    }
  }
  
  return 0;
}
                
void taskQueue_destroy(taskQueue_t tq) {
  pthread_mutex_lock(&(tq->lock));
  tq->isDestroyed = 1;
  int i;
  for (i=0; i < tq->nWorkers; i++) {
    pthread_cond_signal(&(tq->append_or_destroy_cond));
  }
  pthread_mutex_unlock(&(tq->lock));
  int t;  
  for (t=0; t < tq->nWorkers; t++) {
    pthread_join(tq->worker_threads[t], NULL);
  }
  pthread_cond_destroy(&(tq->append_or_destroy_cond));
  pthread_mutex_destroy(&(tq->lock));
  free(tq->worker_threads);
  queue_destroy(tq->q,NULL);
  taskQueue_t_release(tq,NULL);
}

int taskQueue_addTask(taskQueue_t tq, taskQueue_task_fn_t f, taskQueue_closure_t c, taskQueue_taskDone_fn_t doneFunc) {
  pthread_mutex_lock(&(tq->lock));
  int result;
  if (!tq->isDestroyed) {
    taskDescriptor_t td = taskDescriptor_t_acquire();
    td->f = f;
    td->c = c;
    td->doneFn = doneFunc;
    queue_append(tq->q, td);
    if (!tq->nWorkers){
       taskQueueworkerFn(&tq);     
    }
    
    result = 0;
  } else {
    result = 1;
  }
  pthread_cond_signal(&(tq->append_or_destroy_cond));
  pthread_mutex_unlock(&(tq->lock)); 
  return result;
}
